package com.youxin.dataStream.state;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

/**
 * richFunction  状态管理
 */
public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Long>> {
    /**
     * valueState 状态句柄，第一个值为count，第二个为sum值
     * transient 序列化的时候不需要进行序列化，这个由flink框架管理
     */
    private transient ValueState<Tuple2<Long,Long>> sum;

    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
        //获取当前值
        Tuple2<Long, Long> currentSum = sum.value();

        //更新
        currentSum.f0 += 1;
        currentSum.f1 +=input.f1;

        sum.update(currentSum);

        //如果count >=2 清空状态值  重新计算
        if(currentSum.f0 >=2){
            out.collect(new Tuple2(input.f0,currentSum.f1 / currentSum.f0));
            sum.clear();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态
        ValueStateDescriptor descriptor = new ValueStateDescriptor<Tuple2<Long,Long>>("average",//状态名
                TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}),//状态类型
                Tuple2.of(0L,0L));//状态默认值
        sum = getRuntimeContext().getState(descriptor);
    }
}
